package org.zhadoop.streaming.operator;

import java.util.List;
import java.util.Map;

import org.zhadoop.streaming.application.DistributeType;
import org.zhadoop.streaming.config.StreamingConfig;
import org.zhadoop.streaming.event.IEventType;
import org.zhadoop.streaming.exception.StreamingException;
import org.zhadoop.streaming.exception.StreamingRuntimeException;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

/**
 * 基础流处理算子实现类
 * 只实现了基本并发设置，参数设置等方法
 * <p/>
 * Streaming 内部实现都依赖于此类
 * 外部Storm相关不感知此类
 * 
 * @author zwd
 *
 */
public abstract class AbsOperator implements IRichOperator{

	/**
	 * 
	 */
	private static final long serialVersionUID = 4228177169621945974L;
	
	private StreamingConfig conf = null;
	
	private int parallelNumber;
	
	private String operatorId;
	
	private Map<String,GroupInfo> groupInfos;
	
	private Map<String,IEmitter> emitters;
	
	public AbsOperator() {
		groupInfos = Maps.newHashMap(); 
	}
	
	@Override
	public void initialize(Map<String, IEmitter> emitterMap) throws StreamingException {
		this.emitters = emitterMap;
		initialize();
	}

	/**
	 * 初始化
	 * @throws StreamingException
	 */
	public  abstract void initialize() throws StreamingException;
	
	@Override
	public void setConfig(StreamingConfig conf) throws StreamingException {
		this.conf = conf;
	}
	
	@Override
	public StreamingConfig getConfig() {
		return conf;
	}
	
	/**
	 * 设置输入流的名称
	 * @param streamNames
	 * @throws StreamingException
	 */
	public abstract void setInputStream(List<String> streamNames) throws StreamingException;
	
	/**
	 * 设置输出流的名称
	 * @param streamName
	 * @throws StreamingException
	 */
	public abstract void setOutputStream(String streamName) throws StreamingException;
	
	
	/**
	 * 设置输入schema,和输入流的名称一一对应
	 * @param schemas
	 * @throws StreamingException
	 */
	public abstract void setInputSchema(Map<String,IEventType> schemas) throws StreamingException;
	
	/**
	 * 设置输出的Schema
	 * @param schema
	 * @throws StreamingException
	 */
	public abstract void setOutputSchema(IEventType schema) throws StreamingException;

	public int getParallelNumber() {
		return parallelNumber;
	}

	public void setParallelNumber(int parallelNumber) {
		this.parallelNumber = parallelNumber;
	}

	public String getOperatorId() {
		return operatorId;
	}

	public void setOperatorId(String operatorId) {
		this.operatorId = operatorId;
	}

	public Map<String, GroupInfo> getGroupInfos() {
		return groupInfos;
	}

	public void setGroupInfos(Map<String, GroupInfo> groupInfos) {
		this.groupInfos = groupInfos;
	}
	
	/**
	 * 设置算子的分组信息
	 * @param streamName
	 * @param distributeType
	 * @param fields
	 */
	public void setGroupInfo(String streamName,DistributeType distributeType,String[] fields){
		GroupInfo groupInfo = new GroupInfo();
		groupInfo.setStreamName(streamName);
		groupInfo.setDitributeType(distributeType);

		if(null != fields && fields.length > 0){
			//设置分发的时候。如果有多个列，只设置使用一个
			//这也要求，在使用cql的时候，group by 的第一列应该尽可能的分布均匀
			List<String> f = Lists.newArrayList(fields[0]);
			groupInfo.setFields(f);
		}
		
		groupInfos.put(groupInfo.getStreamName(), groupInfo);
		
	}
	
	/**
	 * 
	 * @return
	 */
	public Map<String,IEmitter> getEmitterMap(){
		return emitters;
	}
	
	/**
	 * 通过流名称获取emitter
	 * @param StreamName
	 * @return
	 */
	public IEmitter geEmitter(String streamName){
		if(emitters.containsKey(streamName)){
			return emitters.get(streamName);
		}
		throw new StreamingRuntimeException("can not get emitter by stream name " + streamName);
	}
	
	public IEmitter getEmitter(){
		if(emitters.containsKey(getOutputStream())){
			return emitters.get(getOutputSchema());
		}
		throw new StreamingRuntimeException("can not get emitter by stream name " + this.getOutputStream());
	}
}
